package com.atguigu.marketing.task;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.marketing.bean.MarketingMessage;
import com.atguigu.marketing.bean.MarketingRuleAllInfo;
import com.atguigu.marketing.constant.MarketingConst;
import com.atguigu.marketing.mapper.MarketingRuleMapper;
import com.atguigu.userprofile.utils.RedisUtil;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Component  //组件 spring
public class MarketingTask {

    @Autowired
    MarketingRuleMapper marketingRuleMapper;

    @Autowired
    KafkaTemplate<String,String> kafkaTemplate;

    //      // 1  查询符合的规则列表
    //        // 2 循环列表
    //        /// 2.1  比较时间
    //        //  2.2  查找规则对应的分群
    //        /// 2.3  循环用户
    //        //  2。3.1  检查用户的行为
    //        //  2.3.2   生成消息
    //        //  2.3.3   检查消息限制
    //        //   2.3.4  发送kafka
    @Scheduled(cron = "0/5 * * * * *")  //秒 分 时 天 月 星期
    public void  doTask(){
        Jedis jedis = RedisUtil.getJedis();
        // 1  查询符合的规则列表
        List<MarketingRuleAllInfo>  ruleList=marketingRuleMapper.selectMarketingRuleListForTask();
        System.out.println(ruleList);
        // 2 循环列表
        for (MarketingRuleAllInfo marketingRuleAllInfo : ruleList) {
            /// 2.1  比较时间
           if( ! checkTime(  marketingRuleAllInfo)){
               System.out.println("时间不通过");
               continue;
           }

            System.out.println("时间比较通过");
            //  2.2  查找规则对应的人群集合
            Set<String> userSet = getUserSet(marketingRuleAllInfo, jedis);

            /// 2.3  循环用户
            for (String uid : userSet) {
                //  2。3.1  检查用户的行为
               if(! checkUserAction(  marketingRuleAllInfo ,  uid,  jedis) ){
                   continue;
               }
                // 2.3.2   检查消息限制
                //消息锁    每发送消息时产生消息锁   消息根据时效 解锁  redis expire
                if(!  checkPushLimit(  marketingRuleAllInfo ,  uid,  jedis)){
                    continue;
                }

                //  2.3.3   生成消息
                MarketingMessage marketingMessage = messageBuild(marketingRuleAllInfo, uid);

                //   2.3.4  发送kafka
                kafkaTemplate.send("topic_message" , JSON.toJSONString(marketingMessage));


            }

        }
        jedis.close();
    }

    //@Scheduled(initialDelay = 4000,fixedDelay =2000 )  //计算结束时 到下次计算时的间隔
    public void  doTaskDelay(){
        System.out.println("do task delay ");
    }

   // @Scheduled(fixedRate =2000 )  // 计算开始时 到下次开始时的时间间隔 ，如果下次开始时 上次还没结束 ，则等等直到上次计算结束，下次计算才开始。
    public void  doTaskRate(){
   //     System.out.println("do task rate ");
    }

    private boolean   checkTime(MarketingRuleAllInfo marketingRuleAllInfo){
         if(marketingRuleAllInfo.getPushType().equals("1")) {
             // 发送一次
             Date onceTime = marketingRuleAllInfo.getOnceTime();
             String curTimeStr = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm");
             String onceTimeStr = DateFormatUtils.format(onceTime, "yyyy-MM-dd HH:mm");
             if(curTimeStr.equals(onceTimeStr)){
                 return true;
             }

         }else if (marketingRuleAllInfo.getPushType().equals("2")){  //周期
             if(marketingRuleAllInfo.getLoopType().equals("1")) { //每天

                 String curTimeStr = DateFormatUtils.format(new Date(),   "HH:mm");
                 String loopTimeStr =  marketingRuleAllInfo.getLoopTime();
                 if(curTimeStr.equals(loopTimeStr)){
                     return true;
                 }

             } else if (marketingRuleAllInfo.getLoopType().equals("2")) {//星期
                 String curTimeStr = DateFormatUtils.format(new Date(),   "HH:mm");
                 String curWeekStr = DateFormatUtils.format(new Date(),   "u"); //星期几
                 String loopTimeStr =  marketingRuleAllInfo.getLoopTime();
                 if(curTimeStr.equals(loopTimeStr)&&curWeekStr.equals(marketingRuleAllInfo.getLoopWeekDay()+"")){
                     return true;
                 }
             } else if (marketingRuleAllInfo.getLoopType().equals("3")) {//每月
                 String curTimeStr = DateFormatUtils.format(new Date(),   "HH:mm");
                 String curMonthdayStr = DateFormatUtils.format(new Date(),   "dd"); //每月的日期
                 String loopTimeStr =  marketingRuleAllInfo.getLoopTime();
                 if(curTimeStr.equals(loopTimeStr)&&curMonthdayStr.equals(marketingRuleAllInfo.getLoopMonthDay()+"")){
                     return true;
                 }
             }

         }
         return  false;
    }

    private  Set<String> getUserSet(MarketingRuleAllInfo marketingRuleAllInfo,Jedis jedis){
        List<String> groupIds = marketingRuleAllInfo.getGroupIds();
        String[] userGroupKeys = groupIds.stream().map(id -> "user_group:" + id).toArray(String[]::new);

        Set<String> uidSet = jedis.sunion(userGroupKeys);
        return uidSet;

    }

    private boolean checkUserAction(MarketingRuleAllInfo marketingRuleAllInfo ,String uid,Jedis jedis){
        String key="user_action:uid:"+uid;
        Map<String, String> userActionTodayMap = jedis.hgetAll(key);
        List<JSONObject> userActionConditionList = marketingRuleAllInfo.getUserActionList();
        for (JSONObject userActionConditionJsonObj : userActionConditionList) {
            String userActionNameCondition = userActionConditionJsonObj.getString("userAction");
            Long userActionAmountCondition = userActionConditionJsonObj.getLong("actionAmount");
            String operatorCondition = userActionConditionJsonObj.getString("operator");
            String userActionTodayAmount = userActionTodayMap.getOrDefault(userActionNameCondition, "0");
            boolean compareRs = compare(Long.valueOf(userActionTodayAmount), operatorCondition, userActionAmountCondition);
            if(!compareRs){
                return  false;
            }

        }
        return  true;


    }


    private boolean compare(Long realNum, String opt, Long limitNum) {
        switch (opt) {
            case "gt":
                return realNum.compareTo(limitNum) > 0;
            case "lt":
                return realNum.compareTo(limitNum) < 0;
            case "gte":
                return realNum.compareTo(limitNum) >= 0;
            case "lte":
                return realNum.compareTo(limitNum) <= 0;
            case "eq":
                return realNum.compareTo(limitNum) == 0;
            case "neq":
                return realNum.compareTo(limitNum) != 0;
        }
        return false;
    }

    private  boolean checkPushLimit(MarketingRuleAllInfo marketingRuleAllInfo ,String uid,Jedis jedis){
        //消息锁    每发送消息时产生消息锁   消息根据时效 解锁  redis expire
        //  type  ?  string
        //  key ?   push_lock:[uid]:[rule_id]       value?  "1"
        //  field/score ? 无   写api ?  setex    读api ?   exists     expire?  limitNum+limitUnit => sec
        String lockKey="push_lock:"+uid+":"+marketingRuleAllInfo.getId();
        Integer limitTimeNum = marketingRuleAllInfo.getLimitTimeNum();
        String limitTimeUnit = marketingRuleAllInfo.getLimitTimeUnit();
        Integer expireSec=getSec(limitTimeNum,limitTimeUnit);
        if(!jedis.exists(lockKey)){
            jedis.setex(lockKey,expireSec,"1");
            return  true;
        }else {
            System.out.println("限制发送："+uid);
            return false;
        }


    }


    private Integer getSec(Integer timeNum, String timeUnit  ) {
        switch (timeUnit){
            case MarketingConst.TIME_UNIT_DAY: return timeNum*3600*24 ;
            case MarketingConst.TIME_UNIT_HOUR: return timeNum*3600;
            case MarketingConst.TIME_UNIT_MINIUTE: return timeNum*60;
            case MarketingConst.TIME_UNIT_SEC: return timeNum ;
        }

        return timeNum;
    }

    // 把规则中的消息模板内容 + uid 组合成消息
    public MarketingMessage messageBuild(MarketingRuleAllInfo marketingRuleAllInfo, String uid){
        MarketingMessage marketingMessage = new MarketingMessage();
        marketingMessage.setUrl(marketingRuleAllInfo.getUrl());
        marketingMessage.setText(marketingRuleAllInfo.getModelText());
        marketingMessage.setTitle(marketingRuleAllInfo.getModelTitle());
        marketingMessage.setType(marketingRuleAllInfo.getChannelType());
        marketingMessage.setUid(uid);
        return  marketingMessage;
    }

}
